package cn.azzhu.day06

import cn.azzhu.day02.{SensorReading, SensorSource}
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer
import org.apache.flink.streaming.connectors.redis.RedisSink
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig
import org.apache.flink.streaming.connectors.redis.common.mapper.{RedisCommand, RedisCommandDescription, RedisMapper}

/**
 * @author azzhu
 * @create 2020-09-21 23:42:58
 */
object FlinkWriteToRedis {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val stream = env.addSource(new SensorSource)

    val conf = new FlinkJedisPoolConfig.Builder().setHost("localhost").build()

    stream.addSink(new RedisSink(conf,new MyRedisMapper))

    env.execute("FlinkWriteToKafka")
  }

  class MyRedisMapper extends RedisMapper[SensorReading] {
    override def getCommandDescription: RedisCommandDescription = {
      new RedisCommandDescription(RedisCommand.HSET,"sensor")
    }

    //使用id作为key
    override def getKeyFromData(t: SensorReading): String = t.id

    //使用温度作为value
    override def getValueFromData(t: SensorReading): String = t.temperature.toString
  }
}
